package com.vertx.kafka;

import io.vertx.core.Vertx;
import io.vertx.kafka.client.common.TopicPartition;
import io.vertx.kafka.client.consumer.KafkaConsumer;

import java.util.HashMap;

public class KafkaConsumerTopicOffsetInfo {

    public static void main(String[] args) {

        var config = new HashMap<String, String>();

        config.put("bootstrap.servers", "172.18.30.14:9092");
        config.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        config.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        config.put("group.id", "zhengxing-sync-test.zhengxing.statistic_device-02");
        config.put("auto.offset.reset", "earliest");
        config.put("enable.auto.commit", "false");

        var consumer = KafkaConsumer.create(Vertx.vertx(), config);

        var topic = "zhengxing-sync-test.zhengxing.statistic_device";

        // 订阅topic
        consumer.subscribe(topic)
                .onSuccess(h->{
                    System.out.println("subscribe success");
                })
                .onFailure(e->{
                    System.out.println("subscribe error " + e);
                });
        // 分区处理
        consumer.partitionsAssignedHandler(h->{
            System.out.println("partition assigned");
            h.forEach(tp->{
                System.out.println(tp.getPartition());
            });
            var topicPartition = new TopicPartition(topic, 0);

            // beginningOffsets : 获取分区的第一个offset
            consumer.beginningOffsets(topicPartition)
                    .onSuccess(res->{
                        System.out.println("beginning offset is " + res);
                    });
            // endOffsets  获取分区的最后一个offset
            consumer.endOffsets(topicPartition)
                    .onSuccess(offset->{
                        System.out.println("last commit offset is " + offset);
                    });
            // 定位到起始offset 从这后开始消费消息
            consumer.seekToBeginning(topicPartition)
                    .onSuccess(offset->{
                        System.out.println("seekToBeginning offset is " + offset);
                    })
                    .onFailure(e->{
                        System.out.println("seekToBeginning offset error " + e);
                    });
            // 定位到最后offset
//        consumer.seekToEnd(seekToEndtopicPartition)
//                .onSuccess()
            // beginningOffsets : 获取分区的第一个offset
            consumer.beginningOffsets(topicPartition)
                    .onSuccess(res->{
                        System.out.println("beginning offset is " + res);
                    });
            // endOffsets  获取分区的最后一个offset
            consumer.endOffsets(topicPartition)
                    .onSuccess(offset->{
                        System.out.println("last commit offset is " + offset);
                    });
        })
                .partitionsRevokedHandler(h->{
                    System.out.println("partition revoked");
                    h.forEach(tp->{
                        System.out.println(tp.getPartition());
                    });
                });
    }

}
